package com.bbx.flink.table_api_or_sql.table_api_all;

import com.bbx.flink.demo.entity.Person;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;


@Slf4j
public class SetOperationsDemo {


    private static ExecutionEnvironment executeEnv  = ExecutionEnvironment.getExecutionEnvironment(); ;
    private static BatchTableEnvironment batchTableEnv  = BatchTableEnvironment.create(executeEnv);;
    private static StreamExecutionEnvironment env  = StreamExecutionEnvironment.getExecutionEnvironment();;
    private static StreamTableEnvironment streamTableEnv = StreamTableEnvironment.create(env);



    public static void main(String[] args) throws Exception {
        int flag = 1;

//        DataStream<Person> dataSource1 = env.readTextFile("C:\\Users\\kaifacs\\Desktop\\aaaa.txt").map(i -> {
//            String[] split = i.split(",");
//            return new Person(split[0], split[1], split[2]);
//        });
//        DataStream<Person> dataSource2 = env.readTextFile("C:\\Users\\kaifacs\\Desktop\\bbbb.txt").map(i -> {
//            String[] split = i.split(",");
//            return new Person(split[0], split[1], split[2]);
//        });


        switch (flag) {

            case 1:
//                uniou(dataSource1,dataSource2);

                uniou();
                break;
            case 2:

                break;

            case 3:

                break;


        }

//        env.execute();
        executeEnv.execute();

    }


    /**
     * Similar to a SQL UNION clause. Unions two tables with duplicate records removed.
     * Both tables must have identical field types.
     * 合并，重复的数据去除，注：field  必须具有相同的类型，
     * dataSet1:
     * 111,aa,1
     * 222,bb,2
     * 333,cc,3
     * dataSet2:
     * 333,cc,3
     * result:
     * 111,aa,1
     * 222,bb,2
     * 333,cc,3
     */
    private static void uniou() {
//        DataSet<Person> dataSet1 = executeEnv.readTextFile("C:\\Users\\kaifacs\\Desktop\\aaaa.txt").map(i -> {
//            log.info("~~~~~~~~~~~~~~~i[{}]", i);
//            String[] split = i.split(",");
//            return new Person(split[0], split[1], split[2]);
//        });
//        DataSet<Person> dataSet2 = executeEnv.readTextFile("C:\\Users\\kaifacs\\Desktop\\bbbb.txt").map(i -> {
//            String[] split = i.split(",");
//            return new Person(split[0], split[1], split[2]);
//        });
        DataSource<Person> dataSet1 = executeEnv.fromElements(
                new Person("1", "a", "11"),
                new Person("2", "b", "22"),
                new Person("3", "c", "33")
        );

        DataSource<Person> dataSet2 = executeEnv.fromElements(
                new Person("1", "a", "11"),
                new Person("22", "bb", "222"),
                new Person("3", "c", "33")
        );
        Table leftTable = batchTableEnv.fromDataSet(dataSet1, $("id"),
                $("name"), $("phone"));

        Table rightTable = batchTableEnv.fromDataSet(dataSet2, $("id"),
                $("name"), $("phone"));
        Table unionTable = leftTable.union(rightTable).select($("*"));

        batchTableEnv.toDataSet(unionTable, Person.class).printOnTaskManager("unionTable!!!!");
    }

}
